package com.test.analyse;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.test.common.entity.perf.PerfReportBaseEntity;
import com.test.common.entity.perf.PerfReportEntity;
import com.test.common.entity.perf.inner.ErrorMessageEntity;
import com.test.common.entity.perf.inner.TransactionMessageEntity;
import com.test.common.repository.perf.PerfReportBaseRepository;
import com.test.common.repository.perf.PerfReportRepository;
import com.test.common.utils.ExecutorUtils;
import com.test.common.utils.MD5Utils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.math3.stat.descriptive.DescriptiveStatistics;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.PunctuationType;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map.Entry;
import java.util.Optional;

@Slf4j
public class MyNewProcess implements Processor<String, String> {

	private final PerfReportRepository perfReportRepository;

	private final PerfReportBaseRepository perfReportBaseRepository;

	private int interval = 1;

	// <reportId, <>>
	private HashMap<String, HashMap<String, ArrayList<JSONObject>>> samplers = new HashMap<>();

	private final static double LEVEL_1_PERCENT = 90;
	private final static double LEVEL_2_PERCENT = 95;
	private final static double LEVEL_3_PERCENT = 99;

	public MyNewProcess(int interval, PerfReportRepository perfReportRepository,
			PerfReportBaseRepository perfReportBaseRepository) {

		this.interval = interval;
		this.perfReportRepository = perfReportRepository;
		this.perfReportBaseRepository = perfReportBaseRepository;

		log.info("创建MyNewProcess");
	}

	@Override
	public void init(ProcessorContext context) {

		context.schedule(interval * 1000L, PunctuationType.STREAM_TIME, this::analyse);

		log.info("初始化MyNewProcess");
	}

	/**
	 * 该处方法尽可能简单些
	 *
	 * @param key 报告的id，uuid格式
	 * @param value {"currentHost":"192.168.40.1","delayMS":1000,"errorCount":10,"receiveBytes":10.0,"sampleCount":10,"sampleLabel":"sampleLabel","sentBytes":10.0,"success":true,"threadCount":10,"threadGroupName":"threadGroupName"}
	 */
	@Override
	public void process(String key, String value) {

		log.debug("接受kafka消息：{}  -  {}", key, value);

		HashMap<String, ArrayList<JSONObject>> tempMap = samplers.computeIfAbsent(key, k -> new HashMap<>());

		JSONObject sampleResult = JSON.parseObject(value);

		ArrayList<JSONObject> tmpList = tempMap.computeIfAbsent(sampleResult.getString("sampleLabel"), k -> new ArrayList<>());

		tmpList.add(sampleResult);
	}

	@Override
	public void close() {
	}

	public void analyse(long timeStamp) {

		final HashMap<String, HashMap<String, ArrayList<JSONObject>>> samplersTmp = samplers;
		samplers = new HashMap<>();

		// 此处用上线程池，提高性能
		ExecutorUtils.execute(() -> {

			long start = System.currentTimeMillis();
			long count = 0L;

			for (Entry<String, HashMap<String, ArrayList<JSONObject>>> entry_1 : samplersTmp.entrySet()) {

				log.info("开始解析到报告：{}", entry_1.getKey());

				Optional<PerfReportEntity> optional = perfReportRepository.findByUuid(entry_1.getKey());

				if (!optional.isPresent()) {
					log.error("Kafka中收到不存在ReportId：{}", entry_1.getKey());
					continue;
				}

				PerfReportBaseEntity tmpPerfReportBaseEntity = new PerfReportBaseEntity(entry_1.getKey(), timeStamp);

				// 以下变量用来收集total信息
				int totalSuccessCount = 0;
				int totalFailCount = 0;
				int totalThreads = 0;
				HashMap<String, Integer> totalThreadsMap = new HashMap<>();
				DescriptiveStatistics totalDelayStatistics = new DescriptiveStatistics();
				long totalSendBytes = 0L;
				long totalReceiveBytes = 0L;

				for (Entry<String, ArrayList<JSONObject>> entry_2 : entry_1.getValue().entrySet()) {

					count = count + entry_2.getValue().size();
					log.debug("报告 {} ，Label {}", entry_1.getKey(), entry_2.getKey());

					// 以下变量用来收集各个采样label的信息
					int tempSuccessCount = 0;
					int tempFailCount = 0;
					int tempThreads = 0;
					HashMap<String, Integer> tempThreadsMap = new HashMap<String, Integer>();
					// 用来计算Delay的科学计算
					DescriptiveStatistics tempDelayStatistics = new DescriptiveStatistics();
					long tempSendBytes = 0L;
					long tempReceiveBytes = 0L;

					for (JSONObject sampleResult : entry_2.getValue()) {

						// 单个采样的，只用按照执行机求和，这里不考虑不同线程组中使用同一个label名称的请求
						tempThreadsMap.put(sampleResult.getString("currentHost"), sampleResult.getInteger("threadCount"));
						// 总的求和按照执行机和线程组名称来求和
						totalThreadsMap.put(
								sampleResult.getString("currentHost") + sampleResult.getString("threadGroupName")
										.substring(0, sampleResult.getString("threadGroupName").indexOf("-")),
								sampleResult.getInteger("threadCount"));

						if (sampleResult.getBooleanValue("success")) {
							tempSuccessCount = tempSuccessCount + sampleResult.getIntValue("sampleCount");
							log.debug("成功计数：{}", tempSuccessCount);
						} else {
							log.debug("失败分支");
							tempFailCount = tempFailCount + sampleResult.getIntValue("sampleCount");
							log.debug("失败计数：{}", tempFailCount);

							// 记录错误信息到mongo中
							String responseDataMD5 = MD5Utils.getMD5(sampleResult.getString("responseData"));
							if (tmpPerfReportBaseEntity.getErrorMessageMap().containsKey(responseDataMD5)) {
								tmpPerfReportBaseEntity.getErrorMessageMap().get(responseDataMD5).countUp();
							} else {
								tmpPerfReportBaseEntity.getErrorMessageMap().put(responseDataMD5, new ErrorMessageEntity(
										sampleResult.getString("requestData"), sampleResult.getString("responseData"), 1L));
							}
						}
						tempDelayStatistics.addValue(sampleResult.getIntValue("delayMS"));
						totalDelayStatistics.addValue(sampleResult.getIntValue("delayMS"));
						tempSendBytes = tempSendBytes + sampleResult.getLongValue("sentBytes");
						tempReceiveBytes = tempReceiveBytes + sampleResult.getLongValue("receiveBytes");
					}

					// 计算当前采样label的并发数
					for (Entry<String, Integer> tempEntry : tempThreadsMap.entrySet()) {
						// 集合求和
						tempThreads = tempThreads + tempEntry.getValue();
					}

					// 累计total信息
					totalSuccessCount = totalSuccessCount + tempSuccessCount;
					totalFailCount = totalFailCount + tempFailCount;
					totalSendBytes = totalSendBytes + tempSendBytes;
					totalReceiveBytes = totalReceiveBytes + tempReceiveBytes;

					// 保存各个label事务性能信息
					TransactionMessageEntity tmpTransactionMessageEntity = new TransactionMessageEntity();

					tmpTransactionMessageEntity.setLabel(entry_2.getKey());
					tmpTransactionMessageEntity.setTimeStamp(timeStamp);
					tmpTransactionMessageEntity.setSuccessCount(tempSuccessCount);
					tmpTransactionMessageEntity.setFailCount(tempFailCount);
					tmpTransactionMessageEntity.setQps((tempSuccessCount + tempFailCount) / interval);
					tmpTransactionMessageEntity.setThreads(tempThreads);
					tmpTransactionMessageEntity.setSendBytes(tempSendBytes / interval);
					tmpTransactionMessageEntity.setReceiveByte(tempReceiveBytes / interval);
					tmpTransactionMessageEntity.setLevel_1_Dealy(tempDelayStatistics.getPercentile(LEVEL_1_PERCENT));
					tmpTransactionMessageEntity.setLevel_2_Dealy(tempDelayStatistics.getPercentile(LEVEL_2_PERCENT));
					tmpTransactionMessageEntity.setLevel_3_Dealy(tempDelayStatistics.getPercentile(LEVEL_3_PERCENT));
					tmpTransactionMessageEntity
							.setAvgDealy(tempDelayStatistics.getSum() / (tempSuccessCount + tempFailCount));
					tmpTransactionMessageEntity.setMaxDealy(tempDelayStatistics.getMax());
					tmpTransactionMessageEntity.setMinDealy(tempDelayStatistics.getMin());

					tmpPerfReportBaseEntity.getTransactionMessageList().add(tmpTransactionMessageEntity);

					log.debug("{} - {}", entry_2.getKey(), tmpTransactionMessageEntity.toString());
				}

				// 计算total并发数
				for (Entry<String, Integer> tempEntry : totalThreadsMap.entrySet()) {
					// 集合求和
					totalThreads = totalThreads + tempEntry.getValue();
				}

				// 保存total事务性能信息
				TransactionMessageEntity tmpTransactionMessageEntity = new TransactionMessageEntity();

				tmpTransactionMessageEntity.setLabel("total");
				tmpTransactionMessageEntity.setTimeStamp(timeStamp);
				tmpTransactionMessageEntity.setSuccessCount(totalSuccessCount);
				tmpTransactionMessageEntity.setFailCount(totalFailCount);
				tmpTransactionMessageEntity.setQps((totalSuccessCount + totalFailCount) / interval);
				tmpTransactionMessageEntity.setThreads(totalThreads);
				tmpTransactionMessageEntity.setSendBytes(totalSendBytes / interval);
				tmpTransactionMessageEntity.setReceiveByte(totalReceiveBytes / interval);
				tmpTransactionMessageEntity.setLevel_1_Dealy(totalDelayStatistics.getPercentile(LEVEL_1_PERCENT));
				tmpTransactionMessageEntity.setLevel_2_Dealy(totalDelayStatistics.getPercentile(LEVEL_2_PERCENT));
				tmpTransactionMessageEntity.setLevel_3_Dealy(totalDelayStatistics.getPercentile(LEVEL_3_PERCENT));
				tmpTransactionMessageEntity
						.setAvgDealy(totalDelayStatistics.getSum() / (totalSuccessCount + totalFailCount));
				tmpTransactionMessageEntity.setMaxDealy(totalDelayStatistics.getMax());
				tmpTransactionMessageEntity.setMinDealy(totalDelayStatistics.getMin());

				tmpPerfReportBaseEntity.getTransactionMessageList().add(tmpTransactionMessageEntity);

				log.debug("{total} - {}", tmpTransactionMessageEntity.toString());

				if (optional.get().getPercentAge() < 100) {
					int percentAge = (int) ((timeStamp - optional.get().getStartTime().getTime()) / 10
												/ optional.get().getDuration());
					if (percentAge > 100) {
						log.warn("解析器发现执行进度超过100，{}	{}	{}", optional.get().getId(), optional.get().getTaskId(), optional.get().getScriptId());
						percentAge = 100;
					}
					optional.get().setPercentAge(percentAge);
					perfReportRepository.save(optional.get());
				}
				perfReportBaseRepository.save(tmpPerfReportBaseEntity);

				log.debug("报告：{}，解析结束", entry_1.getKey());
			}

			log.info("解析性能报告数据，数量：{}，耗时：{}", count, System.currentTimeMillis() - start);
		});
	}
}